package com.platform.common.util.avro;

import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.avro.Schema;
import org.apache.avro.file.DataFileReader;
import org.apache.avro.file.DataFileWriter;
import org.apache.avro.file.SeekableByteArrayInput;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.DatumReader;
import org.apache.avro.io.DatumWriter;
import org.apache.avro.specific.SpecificDatumReader;
import org.apache.avro.specific.SpecificDatumWriter;
import org.apache.commons.lang.StringUtils;
import org.apache.log4j.Logger;

public class AvroUtils {

	private static Logger log = Logger.getLogger(AvroUtils.class);
	
	public static List<GenericRecord> deserialization(String schemaName, byte[] array) throws IOException {
		log.debug("反序列化--" + schemaName + "开始");
		if (StringUtils.isEmpty(schemaName)) {
			return new ArrayList<GenericRecord>();
		}
        Schema schema = getSchema(schemaName);
        SeekableByteArrayInput sbai = new SeekableByteArrayInput(array);
        DatumReader<GenericRecord> datumReader = new GenericDatumReader<GenericRecord>(schema);
        DataFileReader<GenericRecord> dataFileReader = new DataFileReader<GenericRecord>(sbai, datumReader);
        List<GenericRecord> list = new ArrayList<GenericRecord>();
        for (GenericRecord genericRecord : dataFileReader) {
            list.add(genericRecord);
        }
        log.debug("反序列化--" + schemaName + "---后生成的数据为---" + list);
        return list;
    }

	public static Schema getSchema(String schemaName) throws IOException {
		String path = AvroUtils.class.getClassLoader().getResource("").getPath();
		log.debug("正在读取" + path + "avro路径下的" + schemaName + ".avsc");
		Schema schema = new Schema.Parser().parse(new File(path + "avro/" + schemaName + ".avsc"));
		return schema;
	}

	/**
	 * TODO 解析二进制流的Avro数据
	 * 
	 * @author zhanghan
	 * @date 2017-3-1
	 * @return 返回解析后的对象
	 * @throws Exception
	 */
	public static <T> T deserialization(Class<T> t, byte[] serializedBytes,
			Schema schema) throws Exception {
		log.debug("反序列化--" + schema.getName() + "开始");
		T clazz = null;
		try {
			SeekableByteArrayInput sbai = new SeekableByteArrayInput(serializedBytes);
			DatumReader<T> reader = new SpecificDatumReader<T>(schema);
			DataFileReader<T> dataFileReader = new DataFileReader<T>(sbai, reader);
			while (dataFileReader.hasNext()) {
				clazz = dataFileReader.next(clazz);
			}
		} catch (IOException e) {
			e.printStackTrace();
		}
		log.debug("反序列化--" + schema.getName() + "---后生成的数据为---" + clazz);
		return clazz;
	}
	
	/**
	 * TODO 解析二进制流的Avro数据
	 * 
	 * @author zhanghan
	 * @date 2017-3-1
	 * @return 返回解析后的对象
	 * @throws Exception
	 */
	@SuppressWarnings({ "unchecked", "rawtypes" })
	public static List deserialization(byte[] serializedBytes, String schemaName)
			throws Exception {
		log.debug("反序列化--" + schemaName + "开始");
		Schema schema = getSchema(schemaName);
		List list = new ArrayList();
		try {
			SeekableByteArrayInput sbai = new SeekableByteArrayInput(
					serializedBytes);
			DatumReader<Object> reader = new SpecificDatumReader<Object>(schema);
			DataFileReader<Object> dataFileReader = new DataFileReader<Object>(sbai, reader);
			for (Object genericRecord : dataFileReader) {
	            list.add(genericRecord);
	        }
		} catch (IOException e) {
			e.printStackTrace();
		}
		log.debug("反序列化--" + schema.getName() + "---后生成的数据为---" + list);
		return list;
	}

	/**
	 * TODO 解析二进制流的Avro数据
	 * 
	 * @author zhanghan
	 * @date 2017-3-1
	 * @return 返回解析后的对象
	 * @throws Exception
	 */
	@SuppressWarnings({ "unchecked", "rawtypes" })
	public static List deserialization(byte[] serializedBytes, Schema schema)
			throws Exception {
		log.debug("反序列化--" + schema.getName() + "开始");
		List list = new ArrayList();
		try {
			SeekableByteArrayInput sbai = new SeekableByteArrayInput(
					serializedBytes);
			DatumReader<Object> reader = new SpecificDatumReader<Object>(schema);
			DataFileReader<Object> dataFileReader = new DataFileReader<Object>(sbai, reader);
			for (Object genericRecord : dataFileReader) {
	            list.add(genericRecord);
	        }
		} catch (IOException e) {
			e.printStackTrace();
		}
		log.debug("反序列化--" + schema.getName() + "---后生成的数据为---" + list);
		return list;
	}

	/**
	 * TODO 解析二进制流的Avro数据
	 * 
	 * @author zhanghan
	 * @date 2017-3-1
	 * @return 返回解析后的对象集合
	 * @throws Exception
	 */
	public static <T> T deserialization(Class<T> t, File file, Schema schema)
			throws Exception {
		if (!file.exists()) {
			return null;
		}
		T object = null;
		try {
			DatumReader<T> datumReader = new SpecificDatumReader<T>(schema);
			DataFileReader<T> dataFileReader = new DataFileReader<T>(file,
					datumReader);
			while (dataFileReader.hasNext()) {
				object = dataFileReader.next();
			}
		} catch (IOException e) {
			e.printStackTrace();
		}
		return object;
	}

	/**
	 * TODO 解析二进制流的Avro数据
	 * 
	 * @author zhanghan
	 * @date 2017-3-1
	 * @return 返回解析后的对象集合
	 * @throws Exception
	 */
	@SuppressWarnings({ "unchecked", "rawtypes" })
	public static List deserialization(File file, Schema schema)
			throws Exception {
		List list = new ArrayList<Object>();
		if (!file.exists()) {
			return list;
		}
		try {
			DatumReader<Object> datumReader = new SpecificDatumReader<Object>(schema);
			DataFileReader<Object> dataFileReader = new DataFileReader<Object>(file, datumReader);
			for (Object genericRecord : dataFileReader) {
	            list.add(genericRecord);
	        }
		} catch (IOException e) {
			e.printStackTrace();
		}
		return list;
	}

	/**
	 * TODO 序列化某个对象到某个文件
	 * 
	 * @author zhanghan
	 * @date 2017-3-1
	 * @param file
	 *            该文件所处的路径必须存在,不然会报错
	 * @return 返回序列化后的文件
	 */
	public static File serialization(Object object, Schema schema, File file)
			throws IOException {
		if (object == null) {
			return null;
		}
		DatumWriter<Object> datumWriter = new SpecificDatumWriter<Object>();
		DataFileWriter<Object> dataFileWriter = new DataFileWriter<Object>(datumWriter);
		dataFileWriter.create(schema, file);
		dataFileWriter.append(object);
		dataFileWriter.flush();
		dataFileWriter.close();
		return file;
	}

	/**
	 * TODO 序列化某个对象到某个文件
	 * 
	 * @author zhanghan
	 * @date 2017-3-1
	 * @param file
	 *            该文件所处的路径必须存在,不然会报错
	 * @return 返回序列化后的文件
	 */
	public static File serialization(List<?> objects, Schema schema, File file)
			throws IOException {
		if (objects == null || objects.size() <= 0) {
			return null;
		}
		DatumWriter<Object> datumWriter = new SpecificDatumWriter<Object>();
		DataFileWriter<Object> dataFileWriter = new DataFileWriter<Object>(datumWriter);
		
		dataFileWriter.create(schema, file);
		for (Object object : objects) {
			dataFileWriter.append(object);
		}
		dataFileWriter.flush();
		dataFileWriter.close();
		return file;
	}

	/**
	 * TODO 序列化某个对象到某个文件
	 * 
	 * @author zhanghan
	 * @date 2017-3-1
	 * @return 返回序列化后的文件二进制数组
	 */
	public static byte[] serialization(Object object, Schema schema)
			throws IOException {
		if (object == null) {
			return null;
		}
		ByteArrayOutputStream out = new ByteArrayOutputStream();
		DatumWriter<Object> writer = new SpecificDatumWriter<Object>(schema);
		DataFileWriter<Object> dataFileWriter = new DataFileWriter<Object>(writer);
		dataFileWriter.create(schema, out);
		dataFileWriter.append(object);
		dataFileWriter.flush();
		dataFileWriter.close();

		byte[] serializedBytes = out.toByteArray();
		return serializedBytes;
	}

	/**
	 * TODO 序列化某个对象到某个文件
	 * 
	 * @author zhanghan
	 * @date 2017-3-1
	 * @param file
	 *            该文件所处的路径必须存在,不然会报错
	 * @return 返回序列化后的文件二进制数组
	 */
	public static byte[] serialization(List<?> objects, Schema schema)
			throws IOException {
		if (objects == null || objects.size() <= 0) {
			return null;
		}
		DatumWriter<Object> writer = new SpecificDatumWriter<Object>(schema);
		ByteArrayOutputStream out = new ByteArrayOutputStream();
		DataFileWriter<Object> dataFileWriter = new DataFileWriter<Object>(writer);
		dataFileWriter.create(schema, out);
		for (Object object : objects) {
			dataFileWriter.append(object);
		}
		dataFileWriter.flush();
		dataFileWriter.close();
		byte[] serializedBytes = out.toByteArray();

		return serializedBytes;
	}

}
